package org.jetlinks.pro.device.message;

import lombok.extern.slf4j.Slf4j;
import org.hswebframework.web.id.IDGenerator;
import org.jetlinks.core.Values;
import org.jetlinks.core.codec.defaults.DeviceMessageCodec;
import org.jetlinks.core.device.DeviceConfigKey;
import org.jetlinks.core.device.DeviceOperator;
import org.jetlinks.core.device.DeviceRegistry;
import org.jetlinks.core.event.EventBus;
import org.jetlinks.core.message.*;
import org.jetlinks.core.server.MessageHandler;
import org.jetlinks.core.server.session.ChildrenDeviceSession;
import org.jetlinks.core.server.session.DeviceSession;
import org.jetlinks.core.server.session.DeviceSessionManager;
import org.jetlinks.core.utils.DeviceMessageTracer;
import org.jetlinks.core.utils.Reactors;
import org.jetlinks.core.utils.StringBuilderUtils;
import org.jetlinks.pro.PropertyConstants;
import org.jetlinks.pro.utils.TopicUtils;
import org.jetlinks.supports.server.DecodedClientMessageHandler;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import javax.annotation.Nonnull;
import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.function.Function;

/**
 * 主要功能:
 * <pre>
 *     1. 实现{@link DecodedClientMessageHandler}统一处理由设备网关解码后的设备消息,将消息转发到事件总线{@link  EventBus}中.
 *     2. 监听设备会话注册注销，生成设备上线离线消息,转发到事件总线中.
 * </pre>
 * <p>
 * 核心方法:
 * {@link  DeviceMessageConnector#onMessage(Message)} ,
 * {@link DeviceMessageConnector#handleMessage(DeviceOperator, Message)}
 *
 * @author zhouhao
 * @see DecodedClientMessageHandler
 * @see TopicUtils
 * @see org.jetlinks.pro.assets.AssetsUtils
 * @see DeviceMessageConnector#onMessage(Message)
 * @since 1.0
 */
@Slf4j
public class DeviceMessageConnector implements DecodedClientMessageHandler, BeanPostProcessor {

    //将设备注册中心的配置追加到消息header中,下游订阅者可直接使用.
    //这些信息是在设备注册时,自动加载到config中的.
    private final static List<String> allConfigHeader = Arrays
        .asList(
            //产品ID
            PropertyConstants.productId.getKey(),
            //设备名称
            PropertyConstants.deviceName.getKey(),
            //租户ID
            PropertyConstants.tenantId.getKey(),
            //租户成员ID
            PropertyConstants.tenantMemberId.getKey(),
            //设备分组ID
            PropertyConstants.groupId.getKey(),
            //机构ID(已过期的配置),已经由bindings替代
            PropertyConstants.orgId.getKey(),
            //绑定信息
            PropertyConstants.bindings.getKey()
        );

    //设备注册中心
    private final DeviceRegistry registry;

    //事件总线
    private final EventBus eventBus;

    //消息处理器,用于将设备回复的指令返回给指令发起者
    private final MessageHandler messageHandler;

    //消息拦截器,用于在消息推送到事件总线之前进行自定义操作,如果添加自定义header等操作
    private final CompositeDeviceMessagePublishInterceptor interceptor = new CompositeDeviceMessagePublishInterceptor();

    //错误监听器,定义全局错误处理
    private final static Function<Throwable, Mono<Void>> doOnError = (error) -> {
        DeviceMessageConnector.log.error(error.getMessage(), error);
        return Mono.empty();
    };

    //空配置定义
    private final static Values emptyValues = Values.of(Collections.emptyMap());

    public DeviceMessageConnector(EventBus eventBus,
                                  DeviceRegistry registry,
                                  MessageHandler messageHandler,
                                  DeviceSessionManager sessionManager) {
        this.registry = registry;
        this.eventBus = eventBus;
        this.messageHandler = messageHandler;

        //从会话管理器里监听会话注册,转发为设备上线消息
        sessionManager
            .onRegister()
            .flatMap(this::handleSessionRegister)
            .subscribe();

        //从会话管理器里监听会话注销,转发为设备离线消息
        sessionManager
            .onUnRegister()
            .flatMap(this::handleSessionUnregister)
            .subscribe();
    }

    protected Mono<Void> handleSessionRegister(DeviceSession session) {
        DeviceOnlineMessage message = new DeviceOnlineMessage();
        message.addHeader("from", "session-register");
        //添加客户端地址信息
        message.addHeader("address", session.getClientAddress().map(InetSocketAddress::toString).orElse(""));
        message.setDeviceId(session.getDeviceId());
        message.setTimestamp(System.currentTimeMillis());
        return this
            .onMessage(message)
            .onErrorResume(doOnError);
    }

    protected Mono<Void> handleSessionUnregister(DeviceSession session) {
        DeviceOfflineMessage message = new DeviceOfflineMessage();
        message.addHeader("from", "session-unregister");
        message.setDeviceId(session.getDeviceId());
        message.setTimestamp(System.currentTimeMillis());
        //子设备会话时添加上级设备id到header中，下游可以直接通过获取header来获取上级设备id
        if (session.isWrapFrom(ChildrenDeviceSession.class)) {
            ChildrenDeviceSession child = session.unwrap(ChildrenDeviceSession.class);
            message.addHeader("parentId", child.getParentDevice().getDeviceId());
        }
        return this
            .onMessage(message)
            .onErrorResume(doOnError);
    }

    public Mono<Void> onMessage(Message message) {
        //标记来忽略处理的消息则忽略处理
        if (null == message || message.getHeader(Headers.ignore).orElse(false)) {
            return Mono.empty();
        }
        //添加消息ID,下游可以使用此ID进行去重
        //场景: 一个消息,可能会被推送到多个topic,比如: 有租户,机构等绑定信息时.
        //     如果用户同时在多个租户(A,B)里,设备也同时在这些租户里,那用户可能会收到多次推送(租户A一次,租户B一次)
        //     这是可以利用uid进行去重,参照: ReactorUtils.distinct
        message.addHeader(PropertyConstants.uid, IDGenerator.SNOW_FLAKE_STRING.generate());
        //添加追踪信息
        DeviceMessageTracer.trace(message, "publish.before");
        return this
            //转化为设备消息
            .convertToDeviceMessage(message)
            .flatMap(msg -> this
                .getTopic(msg)
                //推送到事件总线中
                .flatMap(topic -> eventBus.publish(topic, DeviceMessageCodec.INSTANCE, msg))
                .then())
            //忽略错误
            .onErrorResume(doOnError);
    }

    private Mono<DeviceMessage> convertToDeviceMessage(Message message) {
        //只处理设备消息,其他忽略
        if (message instanceof DeviceMessage) {
            DeviceMessage msg = ((DeviceMessage) message);
            return registry
                .getDevice(msg.getDeviceId())
                //重构消息的header,向header中添加一些常用信息,下游可直接从header中获取使用
                //  而不用再从注册中心里获取
                .flatMap(device -> refactorHeader(device, msg)
                    //执行拦截器
                    .flatMap(newMsg -> interceptor.beforePublish(device, (DeviceMessage) newMsg)))
                .defaultIfEmpty(msg);
        }
        return Mono.empty();
    }

    private Flux<String> getTopic(DeviceMessage message) {
        //构造topic
        Flux<String> topicsStream = Flux.fromIterable(createDeviceMessageTopic(message));
        if (message instanceof ChildDeviceMessage) { //子设备消息
            return this
                //子设备消息也进行转发
                .onMessage(((ChildDeviceMessage) message)
                               .getChildDeviceMessage()
                               .addHeader(DeviceConfigKey.parentGatewayId.getKey(), message.getDeviceId()))
                .thenMany(topicsStream);
        } else if (message instanceof ChildDeviceMessageReply) { //子设备消息回复
            return this
                //子设备消息也进行转发
                .onMessage(((ChildDeviceMessageReply) message)
                               .getChildDeviceMessage()
                               .addHeader(DeviceConfigKey.parentGatewayId.getKey(), message.getDeviceId()))
                .thenMany(topicsStream);
        }
        return topicsStream;
    }

    static Set<String> createDeviceMessageTopic(DeviceMessage deviceMessage) {

        //使用StringBuilderUtils来快速构造topic
        String topic = StringBuilderUtils
            .buildString(deviceMessage, (msg, builder) -> {
                String deviceId = msg.getDeviceId();
                //从header中获取产品ID,
                String productId = msg.getHeaderOrElse(PropertyConstants.productId, () -> "null");
                //topic前缀: /device/{productId}/{deviceId}
                builder.append("/device/")
                       .append(productId)
                       .append("/")
                       .append(deviceId);
                //根据消息拼接topic
                TopicUtils.appendMessageTopic(msg, builder);
            });

        return TopicUtils.refactorTopic(deviceMessage, topic);
    }

    static Mono<Message> refactorHeader(DeviceOperator device, Message message) {
        if (message instanceof DeviceMessage) {
            DeviceMessage deviceMessage = ((DeviceMessage) message);
            String deviceId = deviceMessage.getDeviceId();
            if (StringUtils.isEmpty(deviceId)) {
                log.warn("无法从消息中获取设备ID:{}", deviceMessage);
                return Mono.just(message);
            }
            return device
                .getSelfConfigs(allConfigHeader)
                .defaultIfEmpty(emptyValues)
                .map(configs -> {
                    configs.getAllValues().forEach(deviceMessage::addHeader);
                    return deviceMessage;
                });
        }
        return Mono.just(message);
    }

    public static Flux<String> createDeviceMessageTopic(DeviceRegistry registry, Message message) {
        if (message instanceof DeviceMessage) {
            DeviceMessage deviceMessage = ((DeviceMessage) message);
            String deviceId = deviceMessage.getDeviceId();
            if (StringUtils.isEmpty(deviceId)) {
                log.warn("无法从消息中获取设备ID:{}", deviceMessage);
                return Flux.empty();
            }
            return registry
                .getDevice(deviceId)
                .flatMapMany(device -> refactorHeader(device, deviceMessage)
                    .cast(DeviceMessage.class)
                    .flatMapIterable(DeviceMessageConnector::createDeviceMessageTopic));
        }
        return Flux.just("/device/unknown/message/unknown");
    }

    /**
     * @deprecated 已弃用, 请直接使用: TopicUtils
     */
    @Deprecated
    public static String createDeviceMessageTopic(String productId, String deviceId, DeviceMessage message) {

        return StringBuilderUtils
            .buildString(message, (msg, builder) -> {
                //topic前缀: /device/{productId}/{deviceId}
                builder.append("/device/")
                       .append(productId)
                       .append("/")
                       .append(deviceId);
                //根据消息拼接topic
                TopicUtils.appendMessageTopic(msg, builder);
            });
    }

    protected Mono<Boolean> handleChildrenDeviceMessage(Message message) {
        //消息回复,则返回给指令的发起者
        if (message instanceof DeviceMessageReply) {
            return doReply(((DeviceMessageReply) message));
        }
        //不处理子设备上下线,统一由 DeviceGatewayHelper处理
        return Mono.just(true);
    }

    @Override
    public Mono<Boolean> handleMessage(DeviceOperator device, @Nonnull Message message) {
        Mono<Boolean> then;
        //子设备的消息回复
        if (message instanceof ChildDeviceMessageReply) {
            then = this
                //回复给网关设备
                .doReply(((ChildDeviceMessageReply) message))
                .then(
                    handleChildrenDeviceMessage(((ChildDeviceMessageReply) message).getChildDeviceMessage())
                );
        }
        //子设备消息
        else if (message instanceof ChildDeviceMessage) {
            then = handleChildrenDeviceMessage(((ChildDeviceMessage) message).getChildDeviceMessage());
        }
        //设备回复消息
        else if (message instanceof DeviceMessageReply) {
            then = doReply(((DeviceMessageReply) message));
        }
        //其他消息
        else {
            //do nothing
            then = Reactors.ALWAYS_TRUE;
        }
        return this
            //转发消息
            .onMessage(message)
            .then(then)
            .defaultIfEmpty(false);

    }

    private Mono<Boolean> doReply(DeviceMessageReply reply) {
        if (log.isDebugEnabled()) {
            log.debug("reply message {}", reply.getMessageId());
        }
        //交给处理器统一处理
        return messageHandler
            .reply(reply)
            .thenReturn(true)
            .doOnError((error) -> log.error("reply message error", error))
            ;
    }

    @Override
    public Object postProcessAfterInitialization(@Nonnull Object bean, @Nonnull String beanName) throws BeansException {
        //注册 DeviceMessagePublishInterceptor
        if (bean instanceof DeviceMessagePublishInterceptor) {
            interceptor.addInterceptor(((DeviceMessagePublishInterceptor) bean));
        }
        return BeanPostProcessor.super.postProcessAfterInitialization(bean, beanName);
    }
}
